package com.bff.gaia.streaming.api.mo.nonoptimize;

import com.bff.gaia.api.common.functions.MapPartitionFunction;
import com.bff.gaia.streaming.api.mo.gaia.cache.MaxHeap;
import com.bff.gaia.util.Collector;

import java.util.Iterator;

public class MapPartitionerNon implements MapPartitionFunction<String, Double> {
	private boolean flag = true;
	private MaxHeap maxHeap = null;
	private int N = 200;

	public MapPartitionerNon() {
	}

	public void mapPartition(Iterable<String> values, Collector<Double> out) throws Exception {
		if (this.flag) {
			this.maxHeap = new MaxHeap(this.N);
			this.flag = false;
		}

		Iterator iterator = values.iterator();

		while(iterator.hasNext()) {
			String number = (String)iterator.next();
			if (this.maxHeap.insertElement(Double.valueOf(number))) {
				out.collect(Double.valueOf(number));
			}
		}

	}
}